package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.IndexKlineBean;
import cn.itcast.bean.StockKlineBean;
import cn.itcast.function.IndexMinWindowFunction;
import cn.itcast.function.KeyFunction;
import cn.itcast.function.StockMinWindowFunction;
import cn.itcast.inter.ProcessDataInterface;
import cn.itcast.map.IndexDayKlineMap;
import cn.itcast.util.DorisUtil;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * 指数K线(日,周,月)
 */
public class IndexDayKlineTask implements ProcessDataInterface {
    @Override
    public void process(DataStream<CleanBean> waterData) {
        /**
         * 开发步骤：
         * 1.数据分组
         * 2.划分时间窗口
         * 3.数据处理
         * 4.编写插入sql
         * 数据转换、分组
         * 数据写入mysql
         */
        //2.数据分组
        //6.编写插入sql
        waterData.keyBy(new KeyFunction())
                //3.划分时间窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                //4.数据处理
                .apply(new IndexMinWindowFunction())
                .map(new IndexDayKlineMap())
                //数据分组
                .keyBy(new KeySelector<IndexKlineBean, String>() {
                    @Override
                    public String getKey(IndexKlineBean value) throws Exception {
                        return value.getIndexCode();
                    }
                })
                //数据写入doris
                .addSink(DorisUtil.getSinkFunction("insert into quot_index_kline values(?,?,?,?,?,?,?,?,?,?,?,?,?)"))
                .setParallelism(2);

    }
}
